#ifndef __async_jklkiowefwef89u98uewwefjwejiof23__
#define __async_jklkiowefwef89u98uewwefjwejiof23__

#include <cstdint>
#include <map>
#include <list>
#include <tuple>
#include <functional>
#include <thread>
#include <atomic>
#include <mutex>
#include <condition_variable>
#include <memory>

#include "base_type.hpp"
#include "block_queue.hpp"
#include "log.hpp"

#ifdef _INLINE_
#	undef _INLINE_
#endif

#define _INLINE_ inline

#ifdef ASYNC_THREADS_COUNT
#	define _THREADS_COUNT_     ASYNC_THREADS_COUNT
#else
#	define _THREADS_COUNT_     8U
#endif

/**********************************************************************************************/
namespace hude
{
	/**********************************************************************************************/
	class async_t :
		public std::enable_shared_from_this< async_t >
	{
	public:
		typedef uintptr_t id_t;
		typedef std::chrono::milliseconds interval_t;
		typedef std::chrono::system_clock clock_t;

		enum { emID_NULL = 0 };
		enum : uint8_t { emTYPE_SINGLE, emTYPE_CYCLICITY };

		typedef std::function< void( id_t, uintptr_t ) > on_timer_t;
		typedef std::function< void() > on_task_t;

		typedef clock_t::time_point key_t;

		typedef std::tuple< id_t, uint8_t, interval_t, on_timer_t, uintptr_t > item_t;
		enum { emITEM_ID, emITEM_TYPE, emITEM_INTERVAL, emITEM_CALLBACK, emITEM_USER };

		typedef std::shared_ptr< item_t > ptr_item_t;
		typedef std::shared_ptr< std::thread > ptr_thread_t;

		struct item_comp_t { bool operator()( const key_t& a, const key_t& b ) const { return a < b; } };

		typedef std::multimap< key_t, ptr_item_t, item_comp_t > queue_t;
		typedef std::unique_lock<std::mutex> autolock_t;

		typedef hude::utils::block_queue_t< ptr_item_t > queue_run_t;
		typedef std::list< ptr_thread_t > queue_thread_t;

	public:
		static id_t id();
		static async_t& get_default();
		static std::shared_ptr< async_t > get_default_ptr();

	public:
		async_t( bool run = false, uint_t = _THREADS_COUNT_ );
		~async_t();
		
		bool add( on_task_t );
		bool add( on_timer_t, uintptr_t );
		bool add( id_t, on_timer_t, uintptr_t = 0 );
		bool add( id_t, uint8_t, interval_t, on_timer_t, uintptr_t = 0 );
		bool remove( id_t );
		void clean();
		size_t size() const;

		void start();
		void stop();
		bool running() const;

		std::shared_ptr< async_t > shared();

	private:
		async_t( const async_t& );
		async_t& operator=( const async_t& );
		
		void cb_thread_task();
		void thread_pool( size_t );

		void cb_timer();

	private:
		std::atomic< bool > run_;
		mutable std::mutex mutex_;
		std::condition_variable condition_;
		queue_t queue_;

		std::atomic< size_t > count_run_;
		queue_run_t queue_run_;
		queue_thread_t queue_thread_;

	};

	/**********************************************************************************************/
	_INLINE_ std::shared_ptr< async_t > async_t::get_default_ptr()
	{
		static std::shared_ptr< async_t > async( []{
			auto ptr = new async_t;

			ptr->start();

			return ptr;
		}() );

		return async;
	}

	/**********************************************************************************************/
	_INLINE_ async_t& async_t::get_default()
	{
		return *( get_default_ptr() );
	}

	/**********************************************************************************************/
	_INLINE_ std::shared_ptr< async_t > async_t::shared()
	{
		return this->shared_from_this();
	}

	/**********************************************************************************************/
	_INLINE_ async_t::id_t async_t::id()
	{
		static std::atomic< async_t::id_t > id( 0 );

		id += 1;

		return id;
	}

	/**********************************************************************************************/
	_INLINE_ async_t::async_t( bool run, size_t threads ) : run_( false ), count_run_( threads )
	{
		if( run ) start();
	}
	
	/**********************************************************************************************/
	_INLINE_ async_t::~async_t()
	{
		stop();
	}
	
	/**********************************************************************************************/
	_INLINE_ void async_t::start()
	{
		if( run_ ) return;

		run_ = true;
		
		this->add( 0, std::bind( &async_t::cb_timer, this ), 0 );

		thread_pool( count_run_ );
	}
	
	/**********************************************************************************************/
	_INLINE_ void async_t::stop()
	{
		if( ! run_ ) return;

		run_ = false;

		condition_.notify_all();

		queue_thread_.clear();
	}
	
	/**********************************************************************************************/
	_INLINE_ bool async_t::running() const
	{
		return run_;
	}

	/**********************************************************************************************/
	_INLINE_ bool async_t::add( async_t::id_t id, uint8_t type, interval_t interval, on_timer_t cb_timeout, uintptr_t user )
	{
		autolock_t al( mutex_ );

		for( auto i : queue_ )
		{
			if( std::get< emITEM_ID >( *(i.second) ) == id )
			{
				return false;
			}
		}

		ptr_item_t item( new item_t( id, type, interval, cb_timeout, user ) );

		queue_.insert( std::make_pair(  clock_t::now() + interval, item ) );
		condition_.notify_one();

		return true;
	}
	
	/**********************************************************************************************/
	_INLINE_ bool async_t::add( async_t::id_t id, on_timer_t cb_timeout, uintptr_t user )
	{
		ptr_item_t item( new item_t( id, emTYPE_SINGLE, interval_t( 0 ), cb_timeout, user ) );

		queue_run_.push( item );

		{
			auto c = queue_run_.size();

			if( c > 512U ) log_war( "async_t::queue_run_::size() == %u", c );
		}

		if( count_run_ <= 0 )
		{
			log_inf( "[async_t] new thread from system, idle thread count: %u", (uint32_t)count_run_ );

			// queue_thread_.push_back( ptr_thread_t( new std::thread( &async_t::cb_thread_task, this ),
			// 	[]( std::thread* p ) -> void { p->join(); delete p; } ) );

			std::thread{ &async_t::cb_thread_task, this }.detach();
		}

		return true;
	}
	
	/**********************************************************************************************/
	_INLINE_ bool async_t::add( on_timer_t cb_timeout, uintptr_t user )
	{
		return add( id(), cb_timeout, user );
	}

	/**********************************************************************************************/
	_INLINE_ bool async_t::add( on_task_t cb_task )
	{
		auto f = std::bind( []( on_task_t cb, async_t::id_t, uintptr_t ) { cb(); },
			cb_task, std::placeholders::_1, std::placeholders::_2 );

		return add( id(), f, uintptr_t( 0 ) );
	}

	/**********************************************************************************************/
	_INLINE_ bool async_t::remove( async_t::id_t id )
	{
		autolock_t al( mutex_ );

		for( auto i : queue_ )
		{
			if( std::get< emITEM_ID >( *( i.second ) ) == id )
			{
				queue_.erase( i.first );
				return true;
			}
		}

		return false;
	}

	/**********************************************************************************************/
	_INLINE_ void async_t::clean()
	{
		autolock_t al( mutex_ );
		queue_.clear();
	}

	/**********************************************************************************************/
	_INLINE_ size_t async_t::size() const
	{
		autolock_t al( mutex_ );
		return queue_.size();
	}

	/**********************************************************************************************/
	_INLINE_ void async_t::cb_thread_task()
	{
		ptr_item_t item;

		auto handler = [this]( ptr_item_t& item ) -> void
		{
			count_run_ -= 1;

			// log_dug( "[ timer ] timer( this: %p, count: %d ) BEGIN running task",
			// 	this, (int)count_run_ );

			auto cb = std::get< emITEM_CALLBACK >( *item );
			auto id = std::get< emITEM_ID >( *item );
			auto user = std::get< emITEM_USER >( *item );

			cb( id, user );

			// log_dug( "[ timer ] timer( this: %p, count: %d ) END running task",
			// 	this, (int)count_run_ );

			count_run_ += 1;
		};

		queue_run_.consume( handler, true, std::chrono::milliseconds( 200U ) );
	}

	/**********************************************************************************************/
	_INLINE_ void async_t::thread_pool( size_t siz )
	{
		auto cb = [=]()
		{
			while( run_ )
			{
				cb_thread_task();
			}
		};

		for( size_t i = 0; i < siz; ++i )
		{
			auto thd  = ptr_thread_t( new std::thread( cb ),
				[]( std::thread* p ) -> void { p->join(); delete p; } );

			queue_thread_.push_back( thd );
		}
	}

	/**********************************************************************************************/
	_INLINE_ void async_t::cb_timer()
	{
		interval_t interval( 0 );
		clock_t::time_point now;
		queue_t::iterator itr;
		ptr_item_t item;

		//log_dug( "[ timer ] timer( %p ) start", this );

		while( run_ == true )
		{
			{
				autolock_t al( mutex_ );

				condition_.wait_for( al, interval_t( interval ) );

				if( run_ == false )
				{
					break;
				}

				//log_dug( "[ timer ] timer( %p ) size() = %u", this, queue_.size() );

				if( queue_.empty() )
				{
					itr = queue_.end();
					interval = interval_t( 500 );
					continue;
				} else {
					itr = queue_.begin();
					item = itr->second;
					interval = std::get< emITEM_INTERVAL >( *(item) );
					now = clock_t::now();

					if( now >= itr->first )
					{
						queue_.erase( itr );

						if( std::get< emITEM_TYPE >( *item ) == emTYPE_CYCLICITY )
						{
							itr = queue_.insert( std::make_pair( now + interval, item ) );
							item = itr->second;
						}

						if( item )
						{
							queue_run_.push( item );

							{
								auto c = queue_run_.size();

								if( c > 512U ) log_war( "async_t::queue_run_::size() == %u", c );
							}

							if( count_run_ <= 0 )
							{
								log_inf( "[async_t] new thread from system, idle thread count: %u", (uint32_t)count_run_ );

								// queue_thread_.push_back( ptr_thread_t( new std::thread( &async_t::cb_thread_task, this ),
								// 	[]( std::thread* p ) -> void { p->join(); delete p; } ) );

								std::thread{ &async_t::cb_thread_task, this }.detach();
							}

							//log_dug( "[ timer ] timer( this: %p, size: %u ) tirrger task.", this, queue_.size() );
						}
					} else {
						item.reset();
					}

					if( ! queue_.empty() )
					{
						itr = queue_.begin();

						interval = std::chrono::duration_cast< interval_t >( itr->first - clock_t::now() );

						itr = queue_.end();
					}
				}
			}
		}

		run_ = false;

		//log_dug( "[ timer ] timer( %p ) exit", this );
	}
} //namespace hude

#ifdef _INLINE_
#	undef _INLINE_
#endif

#endif //__async_jklkiowefwef89u98uewwefjwejiof23__
